package com.wxgzh.controller;

import com.wxgzh.service.KafkaService;
import org.apache.http.HttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.CrossOrigin;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.StreamingResponseBody;

import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.attribute.BasicFileAttributes;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;

@RestController
@CrossOrigin
@RequestMapping("/kafka")
public class KafkaController {

    @Autowired
    private KafkaService kafkaService;

    // 定义json文件目录和输出目录
    private static final String JSON_FILE_DIRECTORY = "D:\\MyCoding\\项目库\\sq\\jsonFile\\";
    private static final String OUTPUT_DIRECTORY = "D:\\MyCoding\\项目库\\sq\\output\\";

    // 找到当天创建的文件
    private static File findJsonFileCreatedToday(String directoryPath) {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
        String today = sdf.format(new Date());
        try (Stream<Path> paths = Files.walk(Paths.get(directoryPath))) {
            List<Path> jsonFiles = paths
                    .filter(Files::isRegularFile)
                    .filter(path -> path.toString().endsWith(".json"))
                    .filter(path -> {
                        BasicFileAttributes attrs = null;
                        try {
                            attrs = Files.readAttributes(path, BasicFileAttributes.class);
                        } catch (IOException e) {
                            throw new RuntimeException(e);
                        }
                        return sdf.format(attrs.creationTime().toMillis()).equals(today);
                    })
                    .collect(Collectors.toList());
            if (!jsonFiles.isEmpty()) {
                return jsonFiles.get(0).toFile();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    @GetMapping("/download-today-json")
    public ResponseEntity<StreamingResponseBody> downloadTodayJson(HttpServletResponse response) throws IOException {
        // 获取当天创建的json文件
        File jsonFile = findJsonFileCreatedToday(JSON_FILE_DIRECTORY);
        if (jsonFile == null || !jsonFile.exists()) {
            return ResponseEntity.notFound().build();
        }

        // 设置输出文件路径
        File outputFile = new File(OUTPUT_DIRECTORY + jsonFile.getName());

        // 设置响应头
        HttpHeaders headers = new HttpHeaders();
        headers.setContentType(MediaType.APPLICATION_OCTET_STREAM);
        headers.setContentDispositionFormData("attachment", jsonFile.getName());
        headers.setCacheControl("must-revalidate, post-check=0, pre-check=0");

        // 设置响应体
        StreamingResponseBody responseBody = outputStream -> {
            try (InputStream inputStream = new FileInputStream(jsonFile)) {
                byte[] buffer = new byte[1024];
                int length;
                while ((length = inputStream.read(buffer)) != -1) {
                    outputStream.write(buffer, 0, length);
                }
            }
        };

        // 将文件复制到输出目录
        Files.copy(jsonFile.toPath(), outputFile.toPath());

        return ResponseEntity.ok()
                .headers(headers)
                .body(responseBody);
    }

    @GetMapping("/producer")
    public void producer(){
        kafkaService.kafkaProduceJsonFile();
    }

    @GetMapping("/consumer")
    public void consumer(){
        kafkaService.kafkaConsumeJsonFile();
    }
}